package com.runssnail.demo.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.Properties;
import java.util.concurrent.Future;

/**
 * Hello world!
 */
public class ProducerTest {

    public static final String TOPIC = "topic-person-object";

    public static void main(String[] args) throws Exception {

        KafkaProducer<String, Object> p = null;
        try {
            p = createKafkaProducer(); // new KafkaProducer<String, String>(props);
            Future<RecordMetadata> future = p.send(new ProducerRecord<String, Object>(TOPIC, "test-key",
                    "hello kafka2"));

            // future.wait(2000);

            RecordMetadata meta = future.get();

            System.out.println("partition=" + meta.partition() + ", offset=" + meta.offset());

        } finally {
            if (p != null) {
                p.close();
            }
        }

    }

    public static KafkaProducer<String, Object> createKafkaProducer() {
        Properties props = new Properties();

        // props.put("metadata.broker.list", "localhost:9092");

        // props.put("serializer.class", "kafka.serializer.StringEncoder");
        // props.put("key.serializer.class", "kafka.serializer.StringEncoder");

        props.put("bootstrap.servers", "localhost:9092"); // kafka server
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // key serializer
//        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // value serializer
        props.put("value.serializer", "com.runssnail.demo.kafka.jms.MessageSerializer");

        // request.required.acks
        // 0, which means that the producer never waits for an acknowledgement
        // from the broker (the same behavior as 0.7). This option provides the
        // lowest latency but the weakest durability guarantees (some data will
        // be lost when a server fails).
        // 1, which means that the producer gets an acknowledgement after the
        // leader replica has received the data. This option provides better
        // durability as the client waits until the server acknowledges the
        // request as successful (only messages that were written to the
        // now-dead leader but not yet replicated will be lost).
        // -1, which means that the producer gets an acknowledgement after all
        // in-sync replicas have received the data. This option provides the
        // best durability, we guarantee that no messages will be lost as long
        // as at least one in sync replica remains.
        props.put("request.required.acks", "-1");


        return new KafkaProducer<String, Object>(props);
    }
}
